package com.z.demo

import org.apache.flink.table.api._

/**
 * @Author wenz.ma
 * @Date 2021/10/27 15:17
 * @Desc over()语法，使用RANGE范围
 */
object SqlOverRange01 {
  def main(args: Array[String]): Unit = {
    //创建环境
    val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
    val tEnv = TableEnvironment.create(settings)
    //创建表projects - 使用 kafka connector
    tEnv.executeSql(
      """
        |create table projects(
        |id int,
        |name string,
        |score double,
        |load_date timestamp(3),
        |WATERMARK FOR load_date AS load_date - INTERVAL '1' MINUTE
        |)with(
        |'connector' = 'kafka',
        |'topic' = 'test-topic',
        |'properties.bootstrap.servers' = 'server120:9092',
        |'properties.group.id' = 'testGroup',
        |'scan.startup.mode' = 'latest-offset',
        |'format' = 'csv'
        |)
        |""".stripMargin)
    //加工数据 - 使用over( RANGE )
    tEnv.executeSql(
      """
        |select
        | name,
        | max(score)
        |   over(partition by name
        |     order by load_date
        |     RANGE BETWEEN INTERVAL '10' SECOND PRECEDING AND CURRENT ROW )max_score,
        | min(score)
        |   over(partition by name
        |     order by load_date
        |     RANGE BETWEEN INTERVAL '10' SECOND PRECEDING AND CURRENT ROW )min_score,
        | current_time
        | from
        | projects
        |""".stripMargin).print()//输出到控制台
  }
}
